package com.jiangyg.mall.seckill.service.impl;

import com.alibaba.csp.sentinel.annotation.SentinelResource;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.jiangyg.mall.authz.MemberInfo;
import com.jiangyg.mall.core.support.snowflake.SnowflakeBuilder;
import com.jiangyg.mall.core.utils.Logger;
import com.jiangyg.mall.core.utils.UUIDUtils;
import com.jiangyg.mall.core.utils.bean.BeanUtils;
import com.jiangyg.mall.core.utils.time.DateUtils;
import com.jiangyg.mall.coupon.service.SeckillPromotionService;
import com.jiangyg.mall.coupon.service.SeckillSessionService;
import com.jiangyg.mall.coupon.service.SeckillSkuService;
import com.jiangyg.mall.coupon.vo.SeckillPromotionVO;
import com.jiangyg.mall.coupon.vo.SeckillSessionVO;
import com.jiangyg.mall.coupon.vo.SeckillSkuVO;
import com.jiangyg.mall.order.dto.SeckillOrderDTO;
import com.jiangyg.mall.seckill.constant.CacheConstant;
import com.jiangyg.mall.seckill.dto.SeckillDTO;
import com.jiangyg.mall.seckill.dto.SeckillSessionDetailsDTO;
import com.jiangyg.mall.seckill.dto.SeckillSkuDetailsDTO;
import com.jiangyg.mall.seckill.service.SeckillService;
import com.jiangyg.mall.seckill.vo.SeckillSessionDetailsVO;
import com.jiangyg.mall.seckill.vo.SeckillSkuDetailsVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.ObjectUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.dubbo.config.annotation.Reference;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.*;
import org.springframework.stereotype.Service;

import java.time.*;
import java.util.*;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class SeckillServiceImpl implements SeckillService {

    private final RedissonClient redissonClient;

    private final RabbitTemplate rabbitTemplate;

    private final RedisTemplate<String, Object> redisTemplate;

//    @Reference
    @Autowired
    private SeckillPromotionService seckillPromotionService;

//    @Reference
    @Autowired
    private SeckillSessionService seckillSessionService;

//    @Reference
    @Autowired
    private SeckillSkuService seckillSkuService;

    @Autowired
    public SeckillServiceImpl(RedissonClient redissonClient,
                              RabbitTemplate rabbitTemplate,
                              RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.rabbitTemplate = rabbitTemplate;
        this.redissonClient = redissonClient;
    }

    @Override
    public void updateSeckillDataWithin3Days() throws Exception {
        Logger.info(log, () -> "[上架近3天秒杀数据]-开始执行……");
        // 1. 查询近3天有效的秒杀活动列表
        final List<SeckillPromotionVO> promotionList = seckillPromotionService.selectValidListWithin3Days();
        Logger.info(log, () -> String.format("[上架近3天秒杀数据]-一共查询[%d]个有效的秒杀活动……", promotionList.size()));
        // 2. 遍历所有活动列表，根据场次信息生成秒杀数据
        for (SeckillPromotionVO promotion : promotionList) {
            Logger.info(log, () -> String.format("[上架近3天秒杀数据]-[%s]秒杀活动开始发布……", promotion.getTitle()));
            // 2.1 查询秒杀活动下所有场次列表
            final long promotionId = Long.parseLong(promotion.getId());
            final List<SeckillSessionVO> sessionList = seckillSessionService.selectListByPromotionId(promotionId);
            Logger.info(log, () -> String.format("[上架近3天秒杀数据]-[%s]秒杀活动一共有[%d]场次……", promotion.getTitle(), sessionList.size()));
            // 2.2 如果秒杀活动下还没有设置场次则直接退出本次循环
            if (CollectionUtils.isEmpty(sessionList)) {
                continue;
            }
            // 2.3 生成活动3天内有效的秒杀场次集合
            final List<SeckillSessionDetailsDTO> seckillDetailsList = this.createSeckillDetailsList(promotion, sessionList);
            // 2.4 根据秒杀详情集合发布秒杀数据
            this.publishSeckillSkuBySession(seckillDetailsList, promotion);
            Logger.info(log, () -> String.format("[上架近3天秒杀数据]-[%s]秒杀活动发布完成……", promotion.getTitle()));
        }
        Logger.info(log, () -> "[上架近3天秒杀数据]-执行完毕……");
    }

    /**
     * 功能描述：根据秒杀活动、场次信息生成近3天有效的秒杀详情集合
     *
     * @param promotion   秒杀活动信息
     * @param sessionList 秒杀场次列表
     * @return 近3天有效的秒杀详情集合
     */
    private List<SeckillSessionDetailsDTO> createSeckillDetailsList(SeckillPromotionVO promotion, List<SeckillSessionVO> sessionList) {
        final List<SeckillSessionDetailsDTO> list = new ArrayList<>();
        // 系统默认时区
        ZoneId zoneId = ZoneId.systemDefault();
        // 活动开始相关时间
        final Instant startInstant = promotion.getStartTime().toInstant();
        final LocalDateTime startDateTime = LocalDateTime.ofInstant(startInstant, zoneId);
        // 活动结束相关时间
        final Instant endInstant = promotion.getEndTime().toInstant();
        final LocalDateTime endDateTime = LocalDateTime.ofInstant(endInstant, zoneId);
        final LocalDate endDate = endDateTime.toLocalDate();
        // 定义3天后的时间（生成的最大秒杀场次日期）
        final LocalDate threeAfterDate = LocalDate.now(zoneId).plusDays(3);
        // 当前时间
        LocalDate currentDate = LocalDate.now(zoneId);
        // 循环生成3天内有效的秒杀时间段
        while ((currentDate.isBefore(endDate) || currentDate.isEqual(endDate)) && currentDate.isBefore(threeAfterDate)) {
            for (SeckillSessionVO session : sessionList) {
                // 场次开始时间
                final LocalTime sessionStartTime = session.getStartTime().toLocalTime();
                final LocalDateTime sessionStartDateTime = LocalDateTime.of(currentDate, sessionStartTime);
                final long sessionStartEpochMilli = sessionStartDateTime.toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
                // 如果场次开始时间在活动开始之前则不生成
                if (sessionStartDateTime.isBefore(startDateTime)) {
                    continue;
                }
                // 场次结束时间
                final LocalTime sessionEndTime = session.getEndTime().toLocalTime();
                final LocalDateTime sessionEndDateTime = LocalDateTime.of(currentDate, sessionEndTime);
                final long sessionEndEpochMilli = sessionEndDateTime.toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
                // 生成秒杀详情
                final SeckillSessionDetailsDTO details = new SeckillSessionDetailsDTO();
                details.setId(session.getId());
                details.setName(session.getName());
                details.setStart(sessionStartEpochMilli);
                details.setEnd(sessionEndEpochMilli);
                details.setPid(promotion.getId());
                details.setPtitle(promotion.getTitle());
                list.add(details);
            }
            // 生成下一天的秒杀详情
            currentDate = currentDate.plusDays(1);
        }
        return list;
    }

    /**
     * 功能描述：根据场次信息发布秒杀商品信息
     *
     * @param sessionList 秒杀场次列表
     * @param promotion   秒杀活动
     * @throws Exception 异常
     */
    private void publishSeckillSkuBySession(List<SeckillSessionDetailsDTO> sessionList, SeckillPromotionVO promotion) throws Exception {
        for (SeckillSessionDetailsDTO details : sessionList) {
            Logger.info(log, () -> String.format("[上架近3天秒杀数据]-准备发布秒杀活动[%s]场次[%s]……", details.getPtitle(), details.getName()));
            // 1. 秒杀活动使用 zset 有序集合实现场次的保存
            final String promptionCacheKey = CacheConstant.seckillPromotionKey(details.getPid());
            final BoundZSetOperations<String, Object> zso = redisTemplate.boundZSetOps(promptionCacheKey);
            // 2. 设置活动缓存过期时间（为防止误差额外追加一小时）
            zso.expireAt(DateUtils.addHours(promotion.getEndTime(), 1));
            // 3. 删除已经过期的活动场次（减去10秒减少误差操作）
            zso.removeRangeByScore(0, System.currentTimeMillis() - 10 * 1000);
            // 4. 追加秒杀场次信息（先删后增）
            // TODO 需要测试 object 类型的 remove 是否可以
            zso.remove(details);
            zso.add(details, details.getEnd());
            // 5. 发布秒杀商品的信息
            // 5.1 查询场次下的商品信息
            final long sessionId = Long.parseLong(details.getId());
            final List<SeckillSkuVO> skuList = seckillSkuService.selectListBySessionId(sessionId);
            // 5.2 场次下的商品使用 hash 缓存进行保存

            final String seckillSessionDetailsKey = CacheConstant.seckillSessionKey(details.getPid(), new Date(details.getStart()), sessionId);
            final BoundHashOperations<String, Object, SeckillSkuDetailsDTO> hops = redisTemplate.boundHashOps(seckillSessionDetailsKey);
            // 5.3 遍历商品进行发布
            for (SeckillSkuVO sku : skuList) {
                // 5.3.1 如果秒杀商品已存在缓存则退出本次循环
                if (hops.hasKey(sku.getSkuId())) {
                    continue;
                }
                // 5.3.2 生成秒杀商品的随机码（uuid），秒杀时使用随机码验证，防止刷单
                final String code = UUIDUtils.build();
                // 5.3.3 生成秒杀商品信息并保存缓存
                final SeckillSkuDetailsDTO skuDetails = new SeckillSkuDetailsDTO();
                skuDetails.setCode(code);
                skuDetails.setId(Long.parseLong(sku.getSkuId()));
                skuDetails.setCount(sku.getCount().intValue());
                skuDetails.setLimit(sku.getLimit().intValue());
                skuDetails.setPrice(sku.getPrice());
                skuDetails.setStart(details.getStart());
                skuDetails.setEnd(details.getEnd());
                hops.put(sku.getSkuId(), skuDetails);
                // 5.3.4 设置商品库存（使用信号量处理）
                final String seckillSkuStockKey = CacheConstant.seckillSkuStockKey(code);
                final RSemaphore semaphore = redissonClient.getSemaphore(seckillSkuStockKey);
                semaphore.trySetPermits(skuDetails.getCount().intValue());
            }
        }
    }

    @Override
    public SeckillSessionDetailsVO selectSessionDetails(Long promotionId) throws Exception {
        // 1. 查询当天有效的秒杀场次集合
        final String promptionCacheKey = CacheConstant.seckillPromotionKey(promotionId);
        final BoundZSetOperations<String, Object> zso = redisTemplate.boundZSetOps(promptionCacheKey);
        final long start = System.currentTimeMillis();
        final LocalDateTime end = LocalDateTime.of(LocalDate.now(), LocalTime.MAX);
        final Set<Object> sessions = zso.rangeByScore(start, end.toInstant(ZoneOffset.ofHours(8)).toEpochMilli());
        // TODO 需要验证 set 取出的是否有序（zset有序，主要看下返回的 Set 类型）
        // 2. 取出最小的场次ID，本场次为当前有效的秒杀场次或者下一场秒杀
        if (CollectionUtils.isEmpty(sessions)) {
            return new SeckillSessionDetailsVO();
        }
        final SeckillSessionDetailsDTO dto = (SeckillSessionDetailsDTO) sessions.stream().findFirst().get();
        // 3. 取出场次下所有的商品信息
        final List<SeckillSkuDetailsVO> skuList = this.selectSkuListBySessionId(promotionId, new Date(), Long.parseLong(dto.getId()));
        // 4. 生成返回结果
        return new SeckillSessionDetailsVO(dto.getId(), dto.getName(), skuList);
    }

    /**
     * 功能描述：获取指定场次下商品列表
     *
     * @param promptionId 活动ID
     * @param date        时间
     * @param sessionId   场次ID
     * @return 商品列表
     * @throws Exception 异常
     */
    private List<SeckillSkuDetailsVO> selectSkuListBySessionId(Long promptionId, Date date, Long sessionId) throws Exception {
        final List<SeckillSkuDetailsVO> dataList = new ArrayList<>();
        // 1. 批量从缓存中提取秒杀商品，为防止阻塞，每次提取50条数据
        final ScanOptions options = ScanOptions.scanOptions().count(50).build();
        final String seckillSessionDetailsKey = CacheConstant.seckillSessionKey(promptionId, date, sessionId);
        final BoundHashOperations<String, Object, SeckillSkuDetailsDTO> hops = redisTemplate.boundHashOps(seckillSessionDetailsKey);
        final Cursor<Map.Entry<Object, SeckillSkuDetailsDTO>> cursor = hops.scan(options);
        // 2. 生成商品返回结果
        final long now = System.currentTimeMillis();
        while (cursor.hasNext()) {
            final Map.Entry<Object, SeckillSkuDetailsDTO> entry = cursor.next();
            // 2.1 把 DTO 转成 VO
            final SeckillSkuDetailsVO vo = BeanUtils.transform(entry.getValue(), SeckillSkuDetailsVO.class);
            // 2.2 如果没有到秒杀时刻则随机码置空，防止链接暴露被刷单
            if (vo.getStart() < now || vo.getEnd() > now) {
                vo.setCode(StringUtils.EMPTY);
            }
            dataList.add(vo);
        }
        return dataList;
    }

    // TODO 针对商品进行限流
    @Override
    @SentinelResource(value = "seckill", blockHandler = "seckillBlockHandler")
    public String seckill(SeckillDTO dto, MemberInfo memberInfo) throws InterruptedException {
        // 1. 从缓存取出秒杀商品信息
        final SeckillSkuDetailsDTO details = obtainSkuDetails(dto);
        // 2. 校验秒杀商品参数、商品信息、库存、是否重复购买等
        if (!verifySeckillLegality(details, dto, memberInfo)) {
            return StringUtils.EMPTY;
        }
        // 3. 尝试快速下单（使用分布式信号量先进行库存扣除，如果减扣成功则发送创建订单消息）
        final String seckillSkuStockKey = CacheConstant.seckillSkuStockKey(dto.getCode());
        final RSemaphore semaphore = redissonClient.getSemaphore(seckillSkuStockKey);
        if (semaphore.tryAcquire(dto.getCount(), 100, TimeUnit.MILLISECONDS)) {
            final SeckillOrderDTO order = new SeckillOrderDTO();
            order.setOrderId(SnowflakeBuilder.nextId());
            order.setMemberId(Long.parseLong(memberInfo.getId()));
            order.setPrice(details.getPrice());
            order.setSkuId(details.getId());
            order.setPromotionId(dto.getPromptionId());
            order.setSessionId(dto.getSessionId());
            order.setSubmitDate(new Date());
            rabbitTemplate.convertAndSend("order-event-exchange", "order.seckill.create", order);
            return String.valueOf(order.getOrderId());
        }
        return StringUtils.EMPTY;
    }

    /**
     * 功能描述：获取秒杀商品详情
     *
     * @param dto 参数
     * @return 秒杀商品详情
     */
    private SeckillSkuDetailsDTO obtainSkuDetails(SeckillDTO dto) {
        final String seckillSessionDetailsKey = CacheConstant.seckillSessionKey(dto.getPromptionId(), new Date(), dto.getSessionId());
        final BoundHashOperations<String, Object, SeckillSkuDetailsDTO> hops = redisTemplate.boundHashOps(seckillSessionDetailsKey);
        return hops.get(dto.getSkuId());
    }

    /**
     * 功能描述：验证秒杀合法性
     *
     * @param details    商品详情
     * @param dto        秒杀提交参数
     * @param memberInfo 登录会员信息
     * @return 是否通过
     */
    private boolean verifySeckillLegality(SeckillSkuDetailsDTO details, SeckillDTO dto, MemberInfo memberInfo) {
        // 1. 校验商品是否存在
        if (details == null) {
            return false;
        }
        // 2. 校验是否超过限购数量
        if (details.getLimit() < dto.getCount()) {
            return false;
        }
        // 3. 校验是否到秒杀时间
        long now = System.currentTimeMillis();
        if (details.getStart() > now || details.getEnd() < now) {
            return false;
        }
        // 4. 校验随机码是否合法
        if (StringUtils.isBlank(dto.getCode()) || !StringUtils.equals(dto.getCode(), details.getCode())) {
            return false;
        }
        // 5. 验证库存是否充足
        final int stock = obtainSkuStock(dto);
        if (stock <= 0 || stock < dto.getCount()) {
            return false;
        }
        // 6. 校验是否重复购买（尝试设置会员购买的商品缓存，过期时间：秒杀结束时间 - 当前时间）
        final String seckillPurchaseKey = CacheConstant.seckillPurchaseKey(memberInfo.getId(), dto.getCode());
        long expire = details.getEnd() - System.currentTimeMillis() + 1000;
        return redisTemplate.opsForValue().setIfAbsent(seckillPurchaseKey, dto.getCount(), expire, TimeUnit.MILLISECONDS);
    }

    /**
     * 功能描述：获取商品库存
     *
     * @param dto 参数
     * @return 商品库存
     */
    private int obtainSkuStock(SeckillDTO dto) {
        final String seckillSkuStockKey = CacheConstant.seckillSkuStockKey(dto.getCode());
        final String count = ObjectUtils.toString(redisTemplate.opsForValue().get(seckillSkuStockKey));
        if (NumberUtils.isNumber(count)) {
            return Integer.parseInt(count);
        }
        return 0;
    }

    /**
     * 功能描述：秒杀限流处理
     *
     * @param e 限流异常
     * @return 被限流时返回结果
     */
    public String seckillBlockHandler(BlockException e) {
        // Logger.error(log, () -> "秒杀商品被限流了~~", ex);
        return StringUtils.EMPTY;
    }

}
